package org.example.consumer.push;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.heartbeat.MessageModel;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

/**
 * @author zhangjg
 * @desc c1
 * @date 2023/04/03 09:58
 **/
public class CPush_corrent {

    public static void main(String[] args) throws Exception {

        ExecutorService executorService = Executors.newCachedThreadPool();
        int cnt = 100;
        for (int i = 1; i <= cnt; i++) {
            executorService.submit(new Runnable() {
                @Override
                public void run() {
                    try {
                        // 初始化consumer，并设置consumer group name
                        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("cg_current");
                        // 设置NameServer地址
                        consumer.setNamesrvAddr("127.0.0.1:9876");
                        consumer.setMessageModel(MessageModel.CLUSTERING);
                        //订阅一个或多个topic，并指定tag过滤条件，这里指定*表示接收所有tag的消息
                        consumer.subscribe("t1", "*");
                        //注册回调接口来处理从Broker中收到的消息
                        consumer.registerMessageListener(new MessageListenerConcurrently() {
                            @Override
                            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                                System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), msgs);
                                // 返回消息消费状态，ConsumeConcurrentlyStatus.CONSUME_SUCCESS为消费成功
                                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                            }
                        });
                        // 启动Consumer
                        consumer.start();
                        System.out.println(Thread.currentThread().getName()+" consumer started");
                        System.out.printf("Consumer Started.%n");
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
            });

        }
        System.out.println("----------------job submit end");
        TimeUnit.SECONDS.sleep(120);
        executorService.shutdown();
    }
}
